package operator;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * iterate 算子
 */
public class IterateDemo {

    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> lines = env.socketTextStream("localhost", 9999);

        // map
        SingleOutputStreamOperator<Integer> nums = lines.map(Integer::parseInt);
        // 调用iterate
        IterativeStream<Integer> iterate = nums.iterate();
        // 处理
        SingleOutputStreamOperator<Integer> maped = iterate.map(new MapFunction<Integer, Integer>() {
            @Override
            public Integer map(Integer integer) throws Exception {
                System.out.println("iterate output => " + integer);
                return integer -= 2;
            }
        });
        SingleOutputStreamOperator<Integer> feedback = maped.filter(new FilterFunction<Integer>() {
            @Override
            public boolean filter(Integer integer) throws Exception {
                return integer > 0;
            }
        });
        iterate.closeWith(feedback);

        // 如果不满足条件会重新进行处理
        SingleOutputStreamOperator<Integer> output = maped.filter(new FilterFunction<Integer>() {
            @Override
            public boolean filter(Integer integer) throws Exception {
                return integer <= 0;
            }
        });

        // 打印
        maped.print("output");


        env.execute("iterate");
    }
}
